package data_transfer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;


import java.io.IOException;

/*
    hbase table prepared in hbase:
    create 'hello','cf'
    put 'hello','1','cf:hui','hello world'
    put 'hello','2','cf:hui','hello hadoop'
    put 'hello','3','cf:hui','hello hive'
    put 'hello','4','cf:hui','hello hadoop'
    put 'hello','5','cf:hui','hello world'
    put 'hello','6','cf:hui','hello world'
 */

public class HbaseToHbase {
    private static String DstCF = "words";
    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {
        String SrcHTB = "hello";
        String DstHTB = "cp_hello";
        prepareHTB(DstHTB, DstCF);

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(HbaseToHbase.class);
        job.setJobName("mr_read_write_hbase");

        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(500);

        TableMapReduceUtil.initTableMapperJob(SrcHTB, scan, doMapper.class,
                Text.class, IntWritable.class, job);
        TableMapReduceUtil.initTableReducerJob(DstHTB, doReducer.class, job);
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

    public static void prepareHTB(String tableName, String... tableFamilies) throws IOException {
        // create connection
        Configuration conf = HBaseConfiguration.create();
        Connection conn = ConnectionFactory.createConnection(conf);
        // get admin
        Admin admin = conn.getAdmin();
        TableName tn = TableName.valueOf(tableName);
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tn);
        // add columnFamilies into current tables
        for (String tableFamily : tableFamilies) {
            ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(tableFamily)).build();
            tableDescriptorBuilder.setColumnFamily(cfd);
        }
        TableDescriptor td = tableDescriptorBuilder.build();
        if (admin.isTableAvailable(tn)) {
            System.out.println("table already exists!");
            admin.disableTable(tn);
            admin.deleteTable(tn);
        } else {
            System.out.println("table does not exist!");
            System.out.println("Now creating table ...");
        }
        admin.createTable(td);
    }

    public static class doMapper extends TableMapper<Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            // each value is one Result object, which is one line in hbase storage
            Cell[] cells = value.rawCells();
            String val = null;
            String cf = null;
            String cq = null;
            for (Cell cell : cells) {
                cf = Bytes.toString(CellUtil.cloneFamily(cell));
                cq = Bytes.toString(CellUtil.cloneQualifier(cell));
                val = Bytes.toString(CellUtil.cloneValue(cell));
            }
            context.write(new Text(val), one);
        }
    }

    public static class doReducer extends TableReducer<Text, IntWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, NullWritable, Mutation>.Context context) throws IOException, InterruptedException {
            System.out.println(key.toString());
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            Put put = new Put(Bytes.toBytes(key.toString()));
            put.addColumn(Bytes.toBytes(DstCF),
                    Bytes.toBytes("count"),
                    Bytes.toBytes(String.valueOf(sum)));
            context.write(NullWritable.get(), put);
        }
    }


}
